# @Version        : 1.0
# @Update Time    : 2025/10/15 00:00
# @File           : session.py
# @IDE            : PyCharm
# @Desc           : 会话相关模型
import time
from typing import Any, Dict, Optional, List, Set, Union
from enum import Enum

from pydantic import BaseModel, Field, ConfigDict

from authx.core.event import do_logout_event


class TerminalInfoModel(BaseModel):
    """终端信息模型 - pydantic版本"""

    # 配置模型
    model_config = ConfigDict(from_attributes=True, extra="ignore")

    # 登录会话索引值 (该账号第几个登录的设备, 从 1 开始)
    index: int = 0
    # Token 值
    token_value: Optional[str] = None
    # 所属设备类型，例如：PC、WEB、HD、MOBILE、APP
    device_type: Optional[str] = None
    # 登录设备唯一标识
    device_id: Optional[str] = None
    # 此次登录的自定义扩展数据 (只允许在登录前设定，登录后不建议更改)
    extra_data: Optional[Dict[str, Any]] = None
    # 创建时间（毫秒时间戳）
    create_time: int = Field(default_factory=lambda: int(time.time() * 1000))

    # -------------------- 扩展方法 --------------------

    def set_extra(self, key: str, value: Any) -> "TerminalInfoModel":
        """此次登录的自定义扩展数据 (只允许在登录前设定，登录后不建议更改)"""
        if self.extra_data is None:
            self.extra_data = {}
        self.extra_data[key] = value
        return self

    def get_extra(self, key: str) -> Any:
        """获取扩展数据中的一个值"""
        if self.extra_data is None:
            return None
        return self.extra_data.get(key)

    def have_extra_data(self) -> bool:
        """判断是否设置了扩展数据"""
        return bool(self.extra_data)

    # -------------------- get/set --------------------

    def get_index(self) -> int:
        return self.index

    def set_index(self, index: int) -> "TerminalInfoModel":
        self.index = index
        return self

    def get_token_value(self) -> Optional[str]:
        return self.token_value

    def set_token_value(self, token_value: str) -> "TerminalInfoModel":
        self.token_value = token_value
        return self

    def get_device_type(self) -> Optional[str]:
        return self.device_type

    def set_device_type(self, device_type: str) -> "TerminalInfoModel":
        self.device_type = device_type
        return self

    def get_device_id(self) -> Optional[str]:
        return self.device_id

    def set_device_id(self, device_id: str) -> "TerminalInfoModel":
        self.device_id = device_id
        return self

    def get_extra_data(self) -> Optional[Dict[str, Any]]:
        return self.extra_data

    def set_extra_data(self, extra_data: Dict[str, Any]) -> "TerminalInfoModel":
        self.extra_data = extra_data
        return self

    def get_create_time(self) -> int:
        return self.create_time

    def set_create_time(self, create_time: int) -> "TerminalInfoModel":
        self.create_time = create_time
        return self

    def __str__(self) -> str:
        return (
            "TerminalInfoModel("
            f"index={self.index}, "
            f"token_value='{self.token_value}', "
            f"device_type='{self.device_type}', "
            f"device_id='{self.device_id}', "
            f"extra_data={self.extra_data}, "
            f"create_time={self.create_time}"
            ")"
        )

    # 兼容原有接口
    def to_dict(self) -> Dict[str, Any]:
        return self.model_dump()


# ============= SaSession 主体 =============
# 在 SaSession 上存储数据时使用的键枚举
class SessionKey(str, Enum):
    """
    会话数据存储键枚举，提供类型安全的会话数据访问
    """

    # 用户信息键
    USER = "USER"
    # 角色列表键
    ROLE_LIST = "ROLE_LIST"
    # 权限列表键
    PERMISSION_LIST = "PERMISSION_LIST"


class AuthSessionModel(BaseModel):
    """
    Python 版 SaSession - pydantic版本
    优化Redis交互，提供更强大的数据验证和序列化能力
    """

    # 配置模型
    model_config = ConfigDict(from_attributes=True, extra="ignore")

    # 基本字段
    session_id: Optional[str] = None
    session_type: Optional[str] = None
    login_type: Optional[str] = None
    login_id: Optional[Any] = None
    token: Optional[str] = None
    history_terminal_count: int = 0
    create_time: int = Field(default_factory=lambda: int(time.time() * 1000))

    # 所有挂载数据
    data_map: Dict[str, Any] = Field(default_factory=dict)

    # 登录终端列表
    terminal_list: List[Union[TerminalInfoModel, Dict[str, Any]]] = Field(
        default_factory=list
    )

    def __init__(self, session_id: Optional[str] = None, **kwargs):
        """
        构建一个 Session 对象
        - 与 Java 版一致：如果 session_id 不为 None，则会设置创建时间并发布创建事件
        """
        # 初始化时处理终端列表，确保字典类型的终端数据被转换为TerminalInfoModel
        if "terminal_list" in kwargs:
            processed_terminals = []
            for terminal in kwargs["terminal_list"]:
                if isinstance(terminal, dict):
                    processed_terminals.append(TerminalInfoModel(**terminal))
                else:
                    processed_terminals.append(terminal)
            kwargs["terminal_list"] = processed_terminals

        # 如果提供了session_id，设置创建时间
        if session_id is not None:
            kwargs["create_time"] = int(time.time() * 1000)

        super().__init__(session_id=session_id, **kwargs)

    # ----------------------- 基本字段 get/set -----------------------

    def get_id(self) -> Optional[str]:
        return self.session_id

    def set_id(self, session_id: str) -> "AuthSessionModel":
        self.session_id = session_id
        return self

    def get_type(self) -> Optional[str]:
        return self.session_type

    def set_type(self, session_type: str) -> "AuthSessionModel":
        self.session_type = session_type
        return self

    def get_login_type(self) -> Optional[str]:
        return self.login_type

    def set_login_type(self, login_type: str) -> "AuthSessionModel":
        self.login_type = login_type
        return self

    def get_login_id(self) -> Any:
        return self.login_id

    def set_login_id(self, login_id: Any) -> "AuthSessionModel":
        self.login_id = login_id
        return self

    def get_token(self) -> Optional[str]:
        return self.token

    def set_token(self, token: str) -> "AuthSessionModel":
        self.token = token
        return self

    def get_create_time(self) -> int:
        return self.create_time

    def set_create_time(self, create_time: int) -> "AuthSessionModel":
        self.create_time = create_time
        return self

    # ----------------------- SaTerminalInfo 相关 -----------------------

    def set_terminal_list(
        self, terminal_list: List[Union[TerminalInfoModel, Dict[str, Any]]]
    ):
        # 确保所有终端数据都转换为TerminalInfoModel
        processed_terminals = []
        for terminal in terminal_list:
            if isinstance(terminal, dict):
                processed_terminals.append(TerminalInfoModel(**terminal))
            else:
                processed_terminals.append(terminal)
        self.terminal_list = processed_terminals

    def get_terminal_list(self) -> List[TerminalInfoModel]:
        # 确保返回的都是TerminalInfoModel对象
        return [self._convert_to_terminal_info(t) for t in self.terminal_list]

    def terminal_list_copy(self) -> List[TerminalInfoModel]:
        return list(self.get_terminal_list())

    def get_terminal_list_by_device_type(
        self, device_type: Optional[str]
    ) -> List[TerminalInfoModel]:
        # 返回全部
        if device_type is None:
            return self.get_terminal_list()
        # 返回筛选后的
        return [
            terminal
            for terminal in self.get_terminal_list()
            if terminal.device_type == device_type
        ]

    def _convert_to_terminal_info(self, terminal) -> TerminalInfoModel:
        """将可能是字典的终端数据转换为TerminalInfoModel对象"""
        if isinstance(terminal, TerminalInfoModel):
            return terminal
        elif isinstance(terminal, dict):
            return TerminalInfoModel(**terminal)
        else:
            # 如果是其他类型，尝试创建一个默认的TerminalInfoModel对象
            return TerminalInfoModel(token_value="", device_type="", device_id="")

    def get_token_value_list_by_device_type(
        self, device_type: Optional[str]
    ) -> List[str]:
        token_value_list: List[str] = []
        for terminal in self.get_terminal_list_by_device_type(device_type):
            token_value = terminal.get_token_value()
            if token_value:
                token_value_list.append(token_value)
        return token_value_list

    def get_terminal(self, token_value: str) -> Optional[TerminalInfoModel]:
        for terminal in self.get_terminal_list():
            if terminal.token_value == token_value:
                return terminal
        return None

    async def add_terminal(
        self, terminal_info: Union[TerminalInfoModel, Dict[str, Any]]
    ):
        # 确保terminal_info是TerminalInfoModel对象
        terminal = self._convert_to_terminal_info(terminal_info)

        # 根据 token_value 值查重，如果存在旧的，则先删除
        old_terminal = self.get_terminal(terminal.get_token_value())
        if old_terminal is not None:
            try:
                self.terminal_list.remove(old_terminal)
            except ValueError:
                pass

        # 然后添加新的
        self.history_terminal_count += 1
        terminal.set_index(self.history_terminal_count)
        self.terminal_list.append(terminal)
        await self.update()

    async def remove_terminal(self, token_value: str):
        terminal_info = self.get_terminal(token_value)
        if terminal_info is not None:
            try:
                self.terminal_list.remove(terminal_info)
                await self.update()
            except ValueError:
                pass

    def get_history_terminal_count(self) -> int:
        return self.history_terminal_count

    def set_history_terminal_count(self, history_terminal_count: int):
        self.history_terminal_count = history_terminal_count

    def for_each_terminal_list(self, function: Any):
        """
        遍历 terminal_list 列表，执行特定函数
        - 兼容两种风格：
          1) 具有 run(session, terminal) 方法的对象
          2) 可调用对象：function(session, terminal)
        """
        for terminal_info in self.get_terminal_list():
            if hasattr(function, "run") and callable(function.run):
                function.run(self, terminal_info)
            elif callable(function):
                function(self, terminal_info)
            else:
                raise TypeError(
                    "function 必须是可调用对象，或具有 run(session, terminal) 方法"
                )

    def is_trust_device_id(self, device_id: Optional[str]) -> bool:
        if not device_id:
            return False
        for terminal in self.get_terminal_list():
            if terminal.device_id == device_id:
                return True
        return False

    # ----------------------- 一些操作 -----------------------

    async def update(self):
        """更新Session（从持久库更新刷新一下）"""
        from authx.core.manager import AuthManager

        await AuthManager.get_dao().update_session(self)

    async def logout(self):
        """注销Session (从持久库删除)"""
        if self.session_id is None:
            return
        from authx.core.manager import AuthManager

        await AuthManager.get_dao().delete_session(self.session_id)

        do_logout_event.send(
            "do_logout_event",
            login_type=self.login_type,
            login_id=self.login_id,
            token_value="",
        )

    async def logout_by_terminal_count_to_zero(self):
        """当 Session 上的 SaTerminalInfo 数量为零时，注销会话"""
        if len(self.terminal_list) == 0:
            await self.logout()

    async def timeout(self) -> int:
        """获取此Session的剩余存活时间 (单位: 秒)"""
        if self.session_id is None:
            return 0
        from authx.core.manager import AuthManager

        return await AuthManager.get_dao().get_session_timeout(self.session_id)

    async def update_timeout(self, timeout: int):
        """修改此Session的剩余存活时间 (单位: 秒)"""
        if self.session_id is None:
            return
        from authx.core.manager import AuthManager

        await AuthManager.get_dao().update_session_timeout(self.session_id, timeout)

    async def update_min_timeout(self, min_timeout: int):
        """只有在 Session 的过期时间低于指定的 min_timeout 时才会进行修改"""
        min_v = self.trans(min_timeout)
        curr_v = self.trans(await self.timeout())
        if curr_v < min_v:
            await self.update_timeout(min_timeout)

    async def update_max_timeout(self, max_timeout: int):
        """只有在 Session 的过期时间高于指定的 max_timeout 时才会进行修改"""
        max_v = self.trans(max_timeout)
        curr_v = self.trans(await self.timeout())
        if curr_v > max_v:
            await self.update_timeout(max_timeout)

    def trans(self, value: int) -> int:
        """value为 -1 时返回 Long.MAX_VALUE，否则原样返回"""
        # 延迟导入解决循环引用问题
        from authx.dao.base import AuthTokenDao

        return (1 << 63) - 1 if value == AuthTokenDao.NEVER_EXPIRE else value

    # ----------------------- 存取值 (类型转换) -----------------------

    def get(self, key: str) -> Any:
        """取值"""
        return self.data_map.get(key, None)

    async def set(self, key: str, value: Any) -> "AuthSessionModel":
        """写值"""
        self.data_map[key] = value
        await self.update()
        return self

    async def set_by_null(self, key: str, value: Any) -> "AuthSessionModel":
        """写值 (只有在此 key 原本无值的情况下才会写入)"""
        if key not in self.data_map:
            self.data_map[key] = value
            await self.update()
        return self

    async def delete(self, key: str) -> "AuthSessionModel":
        """删值"""
        if key in self.data_map:
            self.data_map.pop(key, None)
            await self.update()
        return self

    # ----------------------- 其它方法 -----------------------

    def keys(self) -> Set[str]:
        """返回当前 Session 挂载数据的所有 key"""
        return set(self.data_map.keys())

    async def clear(self):
        """清空所有挂载数据"""
        self.data_map.clear()
        await self.update()

    def get_data_map(self) -> Dict[str, Any]:
        """获取数据挂载集合（注意更新 map 里的值后，调用 update() 避免脏数据）"""
        return self.data_map

    def set_data_map(self, data_map: Dict[str, Any]) -> "AuthSessionModel":
        """设置数据挂载集合 (改变底层对象引用，将 data_map 整个对象替换)"""
        self.data_map = data_map
        return self

    async def refresh_data_map(self, data_map: Dict[str, Any]) -> "AuthSessionModel":
        """写入数据集合 (不改变底层对象引用，只将此 data_map 所有数据进行替换)"""
        self.data_map.clear()
        self.data_map.update(data_map)
        await self.update()
        return self

    # ----------------------- Redis优化相关方法 -----------------------

    def to_dict(self) -> Dict[str, Any]:
        """转为字典数据 - 兼容原有接口"""
        return self.model_dump()

    def to_redis_dict(self) -> Dict[str, Any]:
        """专为Redis存储优化的序列化方法"""
        # 使用model_dump可以确保所有数据都被正确序列化
        return self.model_dump(mode="json")

    def model_dump_json_redis(self) -> str:
        """直接返回适合Redis存储的JSON字符串"""
        return self.model_dump_json()

    @classmethod
    def from_redis_dict(cls, data: Dict[str, Any]) -> "AuthSessionModel":
        """从Redis数据创建模型"""
        return cls(**data)

    @classmethod
    def from_redis_json(cls, json_str: str) -> "AuthSessionModel":
        """从Redis存储的JSON字符串创建模型"""
        return cls.model_validate_json(json_str)
